package com.shujia.spark.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator

object Demo25Acc {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo25Acc")

    val sc: SparkContext = new SparkContext(conf)

    var i: Long = 0

    val stuRDD: RDD[String] = sc.textFile("spark/data/stu/students.txt").repartition(10)
    val iAcc: LongAccumulator = sc.longAccumulator

    /**
     * 如果在算子内部对外部的变量进行了修改
     * 实际上修改的是变量的副本
     *
     * 如果需要让内部的修改生效，则可以使用累加器
     */
    stuRDD.foreach(line => {
      i = i + 1
      println(s"i现在的值为：$i")
      // 在每个Task里面独立进行累加
      iAcc.add(1)
      println(s"iAcc现在的值为：${iAcc.value}")
    })

    println(s"foreach算子之后的i现在的值为：$i")
    // 最终会在Driver端进汇总，将每个Task中算好的累加器进行汇总
    println(s"foreach算子之后的iAcc现在的值为：${iAcc.value}")
  }

}
